package cn.doitedu.demo2;

import cn.doitedu.beans.EventBean;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;
/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/14
 * @Desc: 学大数据，上多易教育
 * <p>
静态画像条件（画像库中有现成结果的）：
    职业(tag_01_01)：宝妈，
    年龄(tag_01_02)：20-30
    近半年平均月消费额(tag_03_01)： 1000-1200
动态画像统计条件：
     A行为 3+ 次
     [B行为，E行为 ，Q行为] 序列 2+ 次
规则触发条件：
     X行为

-----------------------------------

静态画像条件（画像库中有现成结果的）：
    职业（tag_01_01）：程序员
    年龄(tag_01_02)：22-28
    兴趣词包含(tag_04_01): 移动硬盘
<p>
动态画像统计条件：
    F行为 2+ 次
    [B行为，D行为 ，K行为] 序列 1+ 次

规则触发条件：
    M行为

 *
 *
 **/
public class RuleCalculator_01 implements RuleCalculator{

    RestHighLevelClient client;
    SearchRequest request;
    BoolQueryBuilder boolQueryBuilder;

    ValueState<Integer> cntState;

    ValueState<Integer> idxState;
    ValueState<Integer> seqCntState;

    String[] actions = {"B", "E", "Q"};

    JSONObject message;



    @Override
    public void init(RuntimeContext runtimeContext) {


        // 先构造客户端对象
        client = new RestHighLevelClient(RestClient.builder(new HttpHost("doitedu", 9200, "http")));


        // 构造请求对象
        request = new SearchRequest("profile_test");


        // 构造职业标签查询条件
        MatchQueryBuilder job = QueryBuilders.matchQuery("tag_01_01", "宝妈");
        // 构造年龄标签查询条件
        RangeQueryBuilder age = QueryBuilders.rangeQuery("tag_01_02").gte(20).lte(30);
        // 构造月平均消费额标签查询条件
        RangeQueryBuilder cost = QueryBuilders.rangeQuery("tag_03_01").gte(1000).lte(1200);


        boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.must(job).must(age).must(cost);

        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-1-1-cnt", Integer.class));

        idxState = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-1-1-idx", Integer.class));
        seqCntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-1-1-seq", Integer.class));

        message = new JSONObject();
        message.put("rule_id","rule-1-1");


    }

    @Override
    public void calc(EventBean eventBean, Collector<String> out) throws IOException {


        long userId = eventBean.getUser_id();
        String eventId = eventBean.getEvent_id();

        /**
         * 一、 查询es，判断当前用户是否满足本规则的静态画像条件
         */

        IdsQueryBuilder id = QueryBuilders.idsQuery().addIds(String.valueOf(userId));

        boolQueryBuilder.must(id);
        request.source(new SearchSourceBuilder().query(boolQueryBuilder));

        // 用客户端发起请求
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);

        // 解析结果
        SearchHits hits = response.getHits();
        // 取出返回的结果条数
        long value = hits.getTotalHits().value;
        // 如果根据上述的条件组合查不到任何结果，则说明当前用户不满足规则的静态画像条件
        if (value < 1) return;

        System.out.println("es查询通过.................");


        /**
         * 二、 对规则的动态画像统计条件进行处理
         */
        if ("A".equals(eventId)) {
            cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);

            System.out.println("更新了A的次数： ................." + cntState.value());
        }

        // String[] actions = {"B", "E", "Q"};
        int idx = idxState.value() == null ? 0 : idxState.value();
        if(eventId.equals(actions[idx])){
            idx++;
            if(idx == actions.length ){
                idx = 0;
                seqCntState.update(seqCntState.value() == null ? 1 : seqCntState.value() + 1);

                System.out.println("更新了序列的次数： ................." + seqCntState.value());


            }
            idxState.update(idx);

        }

        /**
         * 三、 判断当前行为是否是本规则的触发条件事件
         */
        if("X".equals(eventId)){

            System.out.println("触发行为到达：此时的状态数据: " + cntState.value() + "," + seqCntState.value());

            if(cntState.value() != null && cntState.value() >= 3 && seqCntState.value() != null && seqCntState.value() >= 2){

                message.put("user_id",userId);
                message.put("hit_time",eventBean.getAction_time());

                out.collect(message.toJSONString());
            }

        }


    }

    @Override
    public String getRuleId() {
        return "rule-01";
    }
}
